package cn.wangjie.spark.report

import java.sql.{Connection, DriverManager, PreparedStatement}

import cn.wangjie.spark.config.ApplicationConfig
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}

/**
 * 报表开发：按照地域维度（省份和城市）分组统计广告被点击次数
 * 地域分布统计：region_stat_analysis
 */
object RegionStateReport {
	
	/**
	 * 不同业务报表统计分析时，两步骤：
	 *      i. 编写SQL或者DSL分析
	 *      ii. 将分析结果保存MySQL数据库表中
	 */
	def doReport(dataframe: DataFrame): Unit = {
		// 获取SparkSession对象
		val spark: SparkSession = dataframe.sparkSession
		import spark.implicits._
		
		// step1. 将DataFrame注册为临时视图
		dataframe.createOrReplaceTempView("tmp_view_pmt")
		// step2. 编写SQL分析
		val resultDF: DataFrame = spark.sql(
			"""
			  |SELECT
			  |  CAST(DATE_SUB(NOW(), 1)  AS STRING) AS report_date,
			  |  province, city, COUNT(1) AS count
			  |FROM
			  |  tmp_view_pmt
			  |GROUP BY
			  |   province, city
			  |ORDER BY
			  |   count DESC
			  |""".stripMargin)
		//resultDF.printSchema()
		//resultDF.show(10, truncate = false)
		
		// Step3. 保存数据至MySQL表中
		//saveResultToMySQL(resultDF)
		resultDF.coalesce(1).foreachPartition(iter => saveToMySQL(iter))
	}
	
	/**
	 * 针对DataFrame中每个分区数据保存至MySQL表汇总
	 */
	def saveToMySQL(iter: Iterator[Row]): Unit = {
		var conn: Connection = null
		var pstmt: PreparedStatement = null
		
		try{
			// step1. 获取连接
			// 加载驱动类
			Class.forName(ApplicationConfig.MYSQL_JDBC_DRIVER)
			// 获取连接
			conn = DriverManager.getConnection(
				ApplicationConfig.MYSQL_JDBC_URL, //
				ApplicationConfig.MYSQL_JDBC_USERNAME, //
				ApplicationConfig.MYSQL_JDBC_PASSWORD //
			)
			
			val insertSQL =
				"""
				  |INSERT INTO
				  |  itcast_ads_report.region_stat_analysis
				  |  (report_date, province, city, count)
				  |VALUES
				  |  (?, ?, ?, ?)
				  |ON DUPLICATE KEY UPDATE
				  |  count=VALUES(count) ;
				  |""".stripMargin
			pstmt = conn.prepareStatement(insertSQL)
			
			// TODO: 设置手动提交
			conn.setAutoCommit(false)
			
			// step2. 批量写入数据至表中
			iter.foreach{row =>
				pstmt.setString(1, row.getAs[String]("report_date"))
				pstmt.setString(2, row.getAs[String]("province"))
				pstmt.setString(3, row.getAs[String]("city"))
				pstmt.setLong(4, row.getAs[Long]("count"))
				// 加入批次
				pstmt.addBatch()
			}
			
			// 批量插入
			pstmt.executeBatch()
			// TODO: 手动提交
			conn.commit()
		}catch {
			case e: Exception => e.printStackTrace()
		}finally {
			// step3. 关闭连接
			if(null != pstmt) pstmt.close()
			if(null != conn) conn.close()
		}
	
	}
	
	
	/**
	 * 将分析报表数据DataFrame保存至MySQL表中
	 */
	def saveResultToMySQL(dataframe: DataFrame): Unit = {
		dataframe
    		.coalesce(1)
    		.write
			// 如果保存模式为Overwrite，当表存在时，先删除，后创建和插入数据，TODO：以前统计结果全部丢失，不可取
    		//.mode(SaveMode.Overwrite)
			// 保存模式为Append，追加结果，不会出现类似Overwrite是的以前结果数据丢失
    		.mode(SaveMode.Append)
    		.format("jdbc")
			.option("driver", ApplicationConfig.MYSQL_JDBC_DRIVER)
			.option("url", ApplicationConfig.MYSQL_JDBC_URL)
			.option("user", ApplicationConfig.MYSQL_JDBC_USERNAME)
			.option("password", ApplicationConfig.MYSQL_JDBC_PASSWORD)
			.option("dbtable", "itcast_ads_report.region_stat_analysis")
    		.save()
		
		/**
		 * TODO：上述采用 APPEND 保存模式保存DataFrame至MySQL表中，当程序运行多次时（实际中存在），出现主键冲突
		 *      所以保存DataFrame数据至MySQL表中，必须自己手下代码，不能使用dataframe.write方式
		 */
	}
}
